/********** producer & consumer program **********
 * Usage: 
 *   two producers & one consumer corperate together to print
 *   out multiple of ten.
 * Created by Li ZHOU, Apr 11
 */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>

#include <sys/wait.h>
#include <sys/ipc.h>
#include <sys/types.h>
#include <limits.h>
#include <errno.h>
#include <sys/stat.h>

#include <sys/sem.h>
#include <sys/shm.h>

#include <stdarg.h>

/**** const value ****/
#define SP_PERM	 ( S_IRUSR | S_IWUSR )						// semaphore permission
#define SHM_PERM (S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP )	// shardMem permission
#define SET_SIZE  5 
#define STATE_SEM  0			// control read & write status
#define BUFF_ONE_REMAIN 1	// shows remaining space in buffer 1
#define BUFF_TWO_REMAIN 2	// shows remaining space in buffer 2
#define AVAIL_IN_ONE 3		// shows available resource in buffer 1
#define AVAIL_IN_TWO 4		// shows available resource in buffer 2

#define BUFF_SIZE 20
#define ITEM_EACH_LINE 10	// number of output item in each line 

/***** Semaphore Operation *****/
/* apply for semaphore set */
int get_semid();
/* initialize specified # sem in semaphore set to given value */
int init_semid(int sem_id, int index_sem, int value);
/* remove sem */
int rm_sem(int sem_id);
/* set operation struct */
void set_sem_buf(struct sembuf *s, int sem_num, int op, int flag);

/***** Shared Mem Operation *****/
/* apply for share mem */
int get_shmid(int size);
/* detach share mem */
int dt_shm(void * ptr);
/* remove share mem */
int rm_shm(int shm_id);

/***** shared memory struct *****/
typedef struct {
	enum states {running = 0, done} crnt_st; // controlled by proc C
	int buff_one [BUFF_SIZE];                // writen by proc A
	int buff_two [BUFF_SIZE];                // writen by proc B
} SHM;
typedef SHM *SHM_PTR;

#define SHM_SIZE sizeof(SHM)

/***** Processes Function *****/
int producer_func(int sem_id, SHM_PTR p, int parameter );

int consumer_func(int sem_id, SHM_PTR p, int limit );

int state_wait(int sem_id);

int state_signal(int sem_id);

int producer_wait(int sem_id, int parameter);

int producer_signal(int sem_id, int parameter);

int consumer_wait(int sem_id);

int consumer_signal(int sem_id, int cmp_result);

/********** Usage Function ***********/
void usage(char * s) {
	fprintf(stderr, "%s [limit]\n", s);
}

/***********************
 * Main Function 
 ***********************
 */
int main(int argc, char * argv[] ) {
    int limit = 0;
	pid_t child_pid;
	int child_exit_st;

    /*** Initialize basic info ***/
	if(argc != 2) {
        usage(argv[0]);
		exit(1);
	}
    
    /*** Test limit value ***/	
	limit = atoi(argv[1]);
	if(limit <= 0) {			// limit should be positive number 
		fprintf(stderr, "Invalid limit number.Provide a positive integer.\n");
		exit(1);
    }

    /*** set ENV for child processes ***/
    int sem_id = get_semid(); 
	if(sem_id == -1) {
		exit(1);
	}
	/* initialize semaphore value */
    if( init_semid(sem_id, STATE_SEM, 1) == -1 ||
	    init_semid(sem_id, BUFF_ONE_REMAIN, 20) == -1 ||
	    init_semid(sem_id, BUFF_TWO_REMAIN, 20) == -1 ||
        init_semid(sem_id, AVAIL_IN_ONE, 0) == -1 ||
	    init_semid(sem_id, AVAIL_IN_TWO, 0) == -1 
	  ) 
	{
        rm_sem(sem_id);
        exit(1);
	}

	/*  SET SHARED MEMORY   */
	SHM_PTR ptr_shm;
    int shm_id = get_shmid( SHM_SIZE );
    if( (ptr_shm = (SHM_PTR)shmat(shm_id, 0 , 0)) == (SHM_PTR)-1 ) {
		rm_sem(sem_id);
        rm_shm(shm_id);
		perror("attach shm failed");
		exit(1);
	}
	fprintf(stderr, "\tAttach shared memory success\n");
	/* initialize status */
	ptr_shm->crnt_st = running;
	memset(ptr_shm->buff_one, 0, sizeof(int)*20);
	memset(ptr_shm->buff_two, 0, sizeof(int)*20);
    
    /* first fork() for producer */
	// Does not process fork() failed !! 
	// Assuming fork() always success;
	int i;
	pid_t child_arr[3];
	for(i = 0; i < 2; i++ ) {
		if((child_pid = fork())== 0) {
			break;
		}
		else {
			child_arr[i] = child_pid;
		}
	}

	/* producer processes */
	if( child_pid == 0) {
		int parameter = (i == 0? 2:5); // first: proc 1; second: proc 2;
		
		if(producer_func(sem_id, ptr_shm, parameter) == -1){	
			dt_shm( ptr_shm );
			fprintf(stderr, "producer %d abnormal exit: -1\n", getpid());
			exit(1);
		}
		dt_shm( ptr_shm );
		exit(0);
	}
	/* second fork for consumer */
	else {
		/* consumer process */
		if( (child_pid = fork()) == 0 ) {
            if(consumer_func(sem_id, ptr_shm, limit) == -1) {
				dt_shm(ptr_shm);
				fprintf(stderr, "consumer %d abnormal exit: -1\n", getpid());
			    exit(1);
			}
			dt_shm( ptr_shm );
			exit(0);
		}
		/* Main process */
		else {
			child_arr[2] = child_pid;
			/*** Waiting for finish ***/
			while(  (child_pid = wait(&child_exit_st )) ) {
                if( (child_pid == -1) && (errno != EINTR) ) {
					break;
				}
                if(WIFEXITED( child_exit_st ) && WEXITSTATUS(child_exit_st) == 1) {
					fprintf(stderr, "Receiving Abnormal Exit from %d\n",
							child_pid);
					for(i = 0; i < 3; i++) {
						if(child_pid != child_arr[i])
							kill(child_arr[i], SIGKILL);
					}
					break;
				}
			}
            /** rm sem and rm shm **/
            rm_sem(sem_id);
			dt_shm(ptr_shm);
			rm_shm(shm_id);
		}
	}
	return 0;
}


/**************************** operation function ******************************/
/******************************************************************************/

int get_semid( ) {
    int sem_id;
    /* get the semaphore id */
    if((sem_id = semget(IPC_PRIVATE, SET_SIZE, SP_PERM)) == -1) {
		perror("semget()");        
		return -1;
	}
	fprintf(stderr, "\tObtain Semaphore.\n");
	return sem_id;
}


int init_semid(int sem_id, int index_sem, int value) {
    if(semctl(sem_id, index_sem, SETVAL, value) < 0) {
		perror("semctl initial value");
		return -1;
	}
	fprintf(stderr, "\tInitialize sem %d to %d success.\n", index_sem, value);
	return 0;
}

int rm_sem(int sem_id) {
    if(semctl(sem_id, 0, IPC_RMID) == -1 ) {
		perror("rm_sem:");
		return -1;
	}
	fprintf(stderr, "\tRemove sem: %d success.\n", sem_id);
	return 0;
}


void set_sem_buf(struct sembuf * s, int sem_num, int op, int flag) {
    s->sem_num = (short) sem_num;  /* # of item in set */
	s->sem_op = op;            /* negative, 0 , positive */
	/** SET UNDO flag **/
	if(flag == 0)
	    s->sem_flg = SEM_UNDO;         /* IPC_NOWAIT, SEM_UNDO .. */
	else
		s->sem_flg = flag;
}


int state_wait(int sem_id) {
    struct sembuf sem_wait[1];
	set_sem_buf(sem_wait, STATE_SEM, -1, 0); 
	if(semop(sem_id, sem_wait, 1) < 0 ) {
        perror("state wait error");
		return -1;
    }
	return 0;
}

int state_signal(int sem_id) {
	struct sembuf sem_sig[1];
	set_sem_buf(sem_sig, STATE_SEM, 1, 0);
	if(semop(sem_id, sem_sig, 1) < 0 ) {
        perror("state SIG error");
		return -1;
    }
	return 0;
}

int producer_wait(int sem_id, int parameter ) {
	struct sembuf sem_wait[1];
	// proc 1
	if(parameter == 2) {
		set_sem_buf(sem_wait, BUFF_ONE_REMAIN, -1, 0);
	}
	// proc 2
	else {
		set_sem_buf(sem_wait, BUFF_TWO_REMAIN, -1, 0);
	}
	if(semop(sem_id, sem_wait, 1) < 0 ) {
		perror("producer wait failed.");
		return -1;
	}	
    return 0;
}

int producer_signal(int sem_id, int parameter) {
    
	struct sembuf sem_sig[1];
	// proc 1: multiple of 2
	if(parameter == 2) {
		set_sem_buf(sem_sig, AVAIL_IN_ONE, 1, 0);
	}
	// proc 2: multiple of 5
	else {
		set_sem_buf(sem_sig, AVAIL_IN_TWO, 1, 0);
	}
	if(semop(sem_id, sem_sig, 1) < 0 ) {
		perror("producer sig failed.");
		return -1;
	}	
    return 0;
}

int consumer_wait(int sem_id) {
    struct sembuf sem_wait[2];
    set_sem_buf(&sem_wait[0], AVAIL_IN_ONE, -1, 0);	
    set_sem_buf(&sem_wait[1], AVAIL_IN_TWO, -1, 0);
	if(semop(sem_id, sem_wait, 2) < 0 ) {
		perror("consumer wait failed.");
		return -1;
	}	
    return 0;
}

int consumer_signal(int sem_id, int cmp_result) {
	struct sembuf sem_sig[2];
	switch (cmp_result) {
	// 1 > 2: release 1 space in buff 2
	case -1:
        set_sem_buf(&sem_sig[0], AVAIL_IN_ONE, 1, 0);
		set_sem_buf(&sem_sig[1], BUFF_TWO_REMAIN, 1, 0);
		break;
	// 1 < 2: release 1 space in buff 1
	case 1:
        set_sem_buf(&sem_sig[0], AVAIL_IN_TWO, 1, 0);
		set_sem_buf(&sem_sig[1], BUFF_ONE_REMAIN, 1, 0);
		break;
	// 1 == 2: release 1 space in each
	default:
		set_sem_buf(&sem_sig[0], BUFF_ONE_REMAIN, 1, 0);
		set_sem_buf(&sem_sig[1], BUFF_TWO_REMAIN, 1, 0);
		break;
	}
	if(semop(sem_id, sem_sig, 2) < 0 ) {
		perror("consumer sig failed.");
		return -1;
	}
	return 0;
}

int get_shmid(int size) {
    int shmid;
	if( (shmid = shmget(getpid(), size , IPC_CREAT | SHM_PERM )) < 0 ) {
		perror("shmget() failed. ");
		return -1;
	}
	fprintf(stderr, "\tObtain shared memory id success.\n");
	return shmid;
}

int dt_shm(void * ptr) {
    if(shmdt(ptr) < 0 ) {
		perror("shmdt()");
		return -1;
	}
	fprintf(stderr, "\tDetach shared memory success.\n");
	return 0;
}

int rm_shm(int shm_id) {
    if(shmctl(shm_id, IPC_RMID, (struct shmid_ds *)0) == -1 ) {
		perror("remove shared mem.");
		return -1; // remove shared memory
	}
	fprintf(stderr, "\tRemove shared memory success.\n");
    return 0;
}


/**************************** processes function ******************************/
/******************************************************************************/

int producer_func( int sem_id, SHM_PTR shm_ptr, int  parameter ) {
	/*** init value ***/
	int round = 1;
	int current_gen_num = 0;
	while(1) {
		//check the current state
        if(state_wait( sem_id ) == -1)
			return -1;
		if( shm_ptr->crnt_st == done ){
			fprintf(stderr, "\tJob done. Exits now.\n");
			state_signal( sem_id );
			return 0;
		}
		if(state_signal( sem_id ) == -1 )
			return -1;
 
		// generate number 
		current_gen_num = (round * parameter);
		
        //different paramter process different buffer
		if( producer_wait(sem_id, parameter) == -1 )
			return -1;
		//circle buffer
		if( parameter == 2) 
			(shm_ptr->buff_one)[(round - 1)%20] = current_gen_num;
		else
			(shm_ptr->buff_two)[(round - 1)%20] = current_gen_num;
		round++;
		if( producer_signal(sem_id, parameter) == -1 )
			return -1;
    }
}
       
int consumer_func(int sem_id, SHM_PTR shm_ptr, int limit ) {
    int pos_one = 0;
	int pos_two = 0;
	int num_one, num_two;
	int new_line = 0;

    fprintf(stderr, "\n\n**** Output of Program: ****\n\n");
	while(1) {
		if( consumer_wait(sem_id) == -1)
			return -1;
		/** Get number from buffer one **/
		num_one = (shm_ptr->buff_one)[pos_one % 20];
		/** Get number from buffer two **/
		num_two = (shm_ptr->buff_two)[pos_two % 20];
		/** If any number reaches the limit number **/
		if(num_one > limit || num_two > limit) {
			if( state_wait(sem_id) == -1)
				return -1;
			/** Set the running state **/
			shm_ptr->crnt_st = done;
			fprintf(stderr, "\n*** Meet limit. Finish job. ***\n\n");
			if(state_signal(sem_id) == -1)
				return -1;
           
			if(consumer_signal(sem_id, 0) == -1)
				return -1;
			return 0;
		}
		/** compare these two numbers **/
		if(num_one > num_two) {
			pos_two++;
			// keep 1 consume 2 => signal to return 1
			if( consumer_signal(sem_id, -1) == -1)
				return -1;

		}
		else if(num_one < num_two) {
			pos_one++;
			// keep 2 consume 1 => signal to return 2
			if( consumer_signal(sem_id, 1) == -1)
				return -1;
		}
		else {
			pos_one++;
			pos_two++;
            if(new_line > 0 && (new_line % ITEM_EACH_LINE) == 0)
				fprintf(stderr, "\n");
			if( consumer_signal(sem_id, 0) == -1)
				return -1;
			fprintf(stderr, "%d\t", num_one);
			new_line++;
		}
	}
}

  








